use std::{
    future::Future,
    mem::{self, MaybeUninit},
    sync::atomic::{AtomicU8, Ordering},
    task::{ready, Context, Poll},
};

use actix_web::{
    body::BoxBody,
    dev::Payload,
    http::header::ContentType,
    web::{BufMut, Bytes, BytesMut},
    FromRequest, HttpRequest, HttpResponse, Responder,
};
use dashmap::DashMap;
use futures_util::StreamExt;
use tokio::sync::oneshot;

use crate::{
    error::{new_err, AppErr, ErrWrap},
    web::cfg::{
        CBOR_REQUEST_MAX_SIZE, HTTP_RESP_HEAD, HTTP_RESP_OK, REMOTE_CMD, REMOTE_EC, REMOTE_SEQ,
        REMOTE_TYPE,
    },
};

pub struct EventManager {
    seq: AtomicU8,
    event_map: DashMap<u16, oneshot::Sender<Event>>,
}

static mut EVENT_MANAGER: MaybeUninit<EventManager> = MaybeUninit::uninit();

#[derive(Debug, Clone)]
pub struct Event {
    pub ctype: u8,
    pub seq: Option<u8>,
    pub cmd: Option<u8>,
    pub ec: Option<u8>,
    pub body: Bytes,
}

pub const EVENT_TYPE_NOTIFY: u8 = 0;
pub const EVENT_TYPE_NOTIFY_ACK: u8 = 1;
pub const EVENT_TYPE_REQ: u8 = 2;
pub const EVENT_TYPE_RES: u8 = 3;
pub const EVENT_TYPE_ACK: u8 = 4;

pub const CMD_NOTIFY_PRODUCT: u8 = 0;

pub fn init() {
    unsafe {
        EVENT_MANAGER.write(EventManager {
            seq: AtomicU8::new(0),
            event_map: DashMap::new(),
        });
    }
}

pub fn get_event_manager() -> &'static EventManager {
    unsafe { EVENT_MANAGER.assume_init_ref() }
}

pub struct EventDestroy {
    r_ctype: u8,
    pub seq: u8,
    ok: bool,
}

impl EventDestroy {
    pub fn ok(&mut self) {
        self.ok = true;
    }
}

impl Drop for EventDestroy {
    fn drop(&mut self) {
        if !self.ok {
            get_event_manager().remove(self.r_ctype, self.seq);
        }
    }
}

impl EventManager {
    pub fn send_result(&self, event: Event) -> Result<(), AppErr> {
        let seq = event.seq.ok_or(new_err("no seq"))?;
        let id = ((event.ctype as u16) << 8) + seq as u16;
        let tx = self.event_map.remove(&id).ok_or(new_err("not tx ? "))?;
        _ = tx.1.send(event);
        Ok(())
    }

    pub fn create_rx(&self, r_ctype: u8) -> (oneshot::Receiver<Event>, EventDestroy) {
        let (tx, rx) = oneshot::channel();
        let seq = self.seq.fetch_add(1, Ordering::Relaxed);

        let id = ((r_ctype as u16) << 8) + seq as u16;
        self.event_map.insert(id, tx);

        let e = EventDestroy {
            r_ctype,
            ok: false,
            seq,
        };
        (rx, e)
    }

    pub fn remove(&self, r_ctype: u8, seq: u8) {
        let id = ((r_ctype as u16) << 8) + seq as u16;
        self.event_map.remove(&id);
    }
}

impl FromRequest for Event {
    type Error = AppErr;
    type Future = EventFut;

    fn from_request(req: &actix_web::HttpRequest, payload: &mut Payload) -> Self::Future {
        EventFut {
            req: req.clone(),
            payload: payload.take(),
            buf: BytesMut::new(),
        }
    }
}

pub struct EventFut {
    req: HttpRequest,
    payload: Payload,
    buf: BytesMut,
}

impl EventFut {
    fn event(&mut self) -> Result<Event, AppErr> {
        let ctype: u8 = self
            .req
            .headers()
            .get(REMOTE_TYPE)
            .ok_or(new_err("no type"))?
            .to_str()
            .wrap()?
            .parse()
            .wrap()?;

        let seq = self.parse_head(REMOTE_SEQ)?;
        let cmd = self.parse_head(REMOTE_CMD)?;
        let ec = self.parse_head(REMOTE_EC)?;

        let buf = mem::take(&mut self.buf).freeze();

        Ok(Event {
            ctype,
            seq,
            cmd,
            ec,
            body: buf,
        })
    }

    fn parse_head(&self, name: &str) -> Result<Option<u8>, AppErr> {
        let v = match self.req.headers().get(name) {
            None => return Ok(None),
            Some(v) => v,
        };
        Ok(Some(v.to_str().wrap()?.parse().wrap()?))
    }
}

impl Future for EventFut {
    type Output = Result<Event, AppErr>;

    fn poll(self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.get_mut();
        loop {
            let item = ready!(this.payload.poll_next_unpin(cx));
            match item {
                None => return Poll::Ready(this.event()),

                Some(Ok(item)) => {
                    if item.len() + this.buf.len() >= CBOR_REQUEST_MAX_SIZE {
                        return Poll::Ready(Err(new_err("large")));
                    } else {
                        this.buf.put_slice(&item);
                    }
                }
                Some(Err(_)) => return Poll::Ready(Err(new_err("payload err"))),
            }
        }
    }
}

impl Responder for Event {
    type Body = BoxBody;

    fn respond_to(self, _req: &HttpRequest) -> HttpResponse {
        let mut resp = HttpResponse::Ok();
        resp.content_type(ContentType::octet_stream());
        resp.insert_header((HTTP_RESP_HEAD, HTTP_RESP_OK));

        let ctype = format!("{}", self.ctype);
        resp.insert_header((REMOTE_TYPE, ctype));

        if let Some(seq) = self.seq {
            let seq = format!("{}", seq);
            resp.insert_header((REMOTE_SEQ, seq));
        }

        if let Some(cmd) = self.cmd {
            let cmd = format!("{}", cmd);
            resp.insert_header((REMOTE_CMD, cmd));
        }

        if let Some(ec) = self.ec {
            let ec = format!("{}", ec);
            resp.insert_header((REMOTE_EC, ec));
        }

        resp.body(self.body)
    }
}
